package org.misty.redis.bloomfilter;

import lombok.Getter;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.misty.redis.client.RedisClient;
import org.misty.redis.util.Util;

import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.Optional;

import static org.misty.redis.util.Util.checkArgument;

/**
 * @author Misty on 2020-05-30
 */
@Slf4j
@ToString(onlyExplicitlyIncluded = true)
public class BloomFilter {
    private final static int HASH_INCREMENT = 0x61c88647;
    private final static String PUT_SCRIPT;
    private final static String EXISTS_SCRIPT;

    public final static long DEFAULT_VALUE_EXPECTED = 10000;
    public final static double DEFAULT_VALUE_FPP = 0.001;

    static {
        PUT_SCRIPT = Util.copyToString(BloomFilter.class.getClassLoader().getResourceAsStream("lua/bloomfilter/put.lua"));
        EXISTS_SCRIPT = Util.copyToString(BloomFilter.class.getClassLoader().getResourceAsStream("lua/bloomfilter/exists.lua"));
    }

    @Getter
    private final RedisClient client;

    @ToString.Include
    private final String hash = Integer.toHexString(System.identityHashCode(this));
    @ToString.Include
    private final long numBits;
    @ToString.Include
    private final int numHashFunctions;
    @Getter
    @ToString.Include
    private final String key;
    @Getter
    @ToString.Include
    private final String host;
    @Getter
    @ToString.Include
    private final int port;

    private final List<String> keys;

    public BloomFilter(RedisClient redisClient, IFilterConfig config) {
        Objects.requireNonNull(redisClient);
        Objects.requireNonNull(config);

        this.client = redisClient;
        this.host = redisClient.getHost();
        this.port = redisClient.getPort();

        long expectedInsertions = Optional.ofNullable(config.getExpectedInsertions()).orElse(DEFAULT_VALUE_EXPECTED);
        checkArgument(expectedInsertions >= 0, "Expected insertions expectedInsertions(%s) must be >= 0", expectedInsertions);

        double fpp = Optional.ofNullable(config.getFpp()).orElse(DEFAULT_VALUE_FPP);
        checkArgument(fpp > 0.0, "False positive probability fpp(%s) must be > 0.0", fpp);
        checkArgument(fpp < 1.0, "False positive probability fpp(%s) must be < 1.0", fpp);

        this.numBits = optimalNumOfBits(expectedInsertions, fpp);
        this.numHashFunctions = optimalNumOfHashFunctions(expectedInsertions, numBits);

        Objects.requireNonNull(config.getRedisKey());
        String redisKey = config.getRedisKey();
        this.key = redisKey;
        this.keys = Collections.singletonList(redisKey);

        log.info("Initialized with {expectedInsertions={}, fpp={}, redisKey={}, numBits={}, numHash={}}",
            expectedInsertions, fpp, redisKey, numBits, numHashFunctions);
    }

    public boolean put(Object key) {
        String[] indexes = getIndexes(key, numBits, numHashFunctions);
        Object ret = client.eval(PUT_SCRIPT, String.class, keys, indexes);
        if (ret != null) {
            return "OK".equals(ret);
        }
        return false;
    }

    public boolean exists(Object key) {
        String[] indexes = getIndexes(key, numBits, numHashFunctions);
        Long ret = client.eval(EXISTS_SCRIPT, Long.class, keys, indexes);
        if (ret != null) {
            return ret.equals(1L);
        }
        return false;
    }

    /**
     * 内部方法
     * 计算索引
     */
    public static String[] getIndexes(Object k, long numBits, int numHashFunctions) {
        String[] idx = new String[numHashFunctions];
        long h = hash(k);
        for (int i = 0; i < numHashFunctions; i++) {
            idx[i] = String.valueOf(Math.abs(h) % numBits);
            h += HASH_INCREMENT;
        }
        return idx;
    }

    public static long hash(Object k) {
        int h = 0;
        if (k instanceof String) {
            h = murmur3_32(0, (String) k);
        } else {
            h ^= k.hashCode();
            h ^= (h >>> 20) ^ (h >>> 12);
            h ^= (h >>> 7) ^ (h >>> 4);
        }
        return h;
    }

    public static int murmur3_32(int seed, String value) {
        byte[] data = value.getBytes();
        int len = data.length;
        int res = seed;
        int pos = 0;
        int temp;
        int i;
        for (i = len; i >= 2; res = res * 5 + -430675100) {
            temp = data[pos++] & '\uffff' | data[pos++] << 16;
            i -= 2;
            temp *= -862048943;
            temp = Integer.rotateLeft(temp, 15);
            temp *= 461845907;
            res ^= temp;
            res = Integer.rotateLeft(res, 13);
        }

        if (i > 0) {
            temp = data[pos] * -862048943;
            temp = Integer.rotateLeft(temp, 15);
            temp *= 461845907;
            res ^= temp;
        }

        res ^= len * 2;
        res ^= res >>> 16;
        res *= -2048144789;
        res ^= res >>> 13;
        res *= -1028477387;
        res ^= res >>> 16;
        return res;
    }

    /*
     * Guava 29.0
     */
    private static long optimalNumOfBits(long n, double p) {
        if (p == 0) {
            p = Double.MIN_VALUE;
        }
        return (long) (-n * Math.log(p) / (Math.log(2) * Math.log(2)));
    }

    /*
     * Guava 29.0
     */
    private static int optimalNumOfHashFunctions(long n, long m) {
        // (m / n) * log(2), but avoid truncation due to division!
        return Math.max(1, (int) Math.round((double) m / n * Math.log(2)));
    }
}
